package oktx.dubbo.interceptor;

import com.alibaba.dubbo.common.Constants;
import com.alibaba.dubbo.common.extension.Activate;
import com.alibaba.dubbo.common.utils.ReflectUtils;
import com.alibaba.dubbo.rpc.*;
import oktx.JtaTransaction;
import oktx.TccTransaction;
import oktx.TransactionContext;
import oktx.dubbo.config.JsonProcessor;
import oktx.dubbo.participant.DubboTccParticipant;
import oktx.participant.JtaParticipant;
import oktx.participant.Participant;
import oktx.RemoteType;

import java.lang.reflect.Method;
import java.util.Arrays;

/**
 * Created by dave on 18-5-18 下午9:04.
 * 拦截Dubbo调用，当前有事务时发起的需要JTA事务支持的Dubbo调用记入当前事务上下文中
 */
@Activate(group = Constants.CONSUMER)
public class DubboConsumerInterceptor implements Filter {

    @Override
    public Result invoke(Invoker<?> invoker, Invocation invocation) throws RpcException {
        TransactionContext context = TransactionContext.getContext();
        if (context != null) {
            Method method;
            try {
                method = ReflectUtils.findMethodByMethodSignature(invoker.getInterface(), invocation.getMethodName(),
                    Arrays.stream(invocation.getParameterTypes()).map(Class::getName).toArray(String[]::new));
            } catch (Exception e) {
                throw new RpcException(e);
            }

            JtaTransaction jta = method.getAnnotation(JtaTransaction.class);
            TccTransaction tcc = method.getAnnotation(TccTransaction.class);
            Participant p = null;
            if (jta != null) {
                String ip = invoker.getUrl().getIp();
                int port = invoker.getUrl().getPort();
                JtaParticipant jp = (JtaParticipant) context.getParticipants().stream().filter(e -> {
                    if (e instanceof JtaParticipant) {
                        JtaParticipant j = (JtaParticipant) e;
                        return j.getType() == RemoteType.dubbo && j.getPort() == port && j.getIp().equals(ip);
                    }
                    return false;
                }).findFirst().orElse(null); // 对于Jta事务，同一微服务实例只加入一次

                if (jp == null) {
                    jp = new JtaParticipant();
                    jp.setXid(context.getXid());
                    jp.setIp(ip);
                    jp.setPort(port);
                    jp.setTimeout(jta.timeout());
                    jp.setType(RemoteType.dubbo);
                    p = jp;
                }
                String json = JsonProcessor.toJson(jp);
                RpcContext.getContext().setAttachment(JtaParticipant.PRC_TRANSACTION_KEY, json);
            } else if (tcc != null) {
                DubboTccParticipant dp = new DubboTccParticipant();
                dp.setXid(context.getXid());
                dp.setClazz(invoker.getInterface());
                dp.setMethod(invocation.getMethodName());
                dp.setConfirmMethod(tcc.confirmMethod());
                dp.setCancelMethod(tcc.cancelMethod());
                dp.setIp(invoker.getUrl().getIp());
                dp.setPort(invoker.getUrl().getPort());
                p = dp;
            }

            Result r = invoker.invoke(invocation);
            if (!r.hasException() && p != null) { // 有异常时,被调用方已自行回滚事务
                context.getParticipants().add(p);
            }
            return r;
        }

        return invoker.invoke(invocation);
    }

}
